use axum::{
    extract::*,
    middleware::{self},
    routing::{get, on, post, MethodFilter},
    Router,
};
use kafka::{
    consumer::{Consumer, FetchOffset, GroupOffsetStorage},
    producer::{Producer, Record, RequiredAcks},
};
use nacos_demo::*;
use std::fmt::Write;
use std::{net::SocketAddr, time::Duration};
use tracing::info;
#[tokio::main]
pub async fn main() {
    log_init();
    let config = load_yaml();
    // let mut _ip = config.server.host.unwrap();
    // register_nacos(config.server.clone()).await;
    let server_config = config.server.clone();
    let port = server_config.port.unwrap();
    let ip = server_config.host;

    let r = platform::init(
        &config.db.username,
        &config.db.password,
        &config.db.host,
        &config.db.port,
        &config.db.database,
    )
    .await;
    init_kafka();

    info!("mybatis init success ");

    let app = Router::new()
        .route("/", get(handler))
        .route("/valid", get(validated_handler))
        .route("/", post(|| async move { "Hello from `POST /`" }))
        .layer(middleware::from_fn(print_request_response))
        .layer(Extension(r));

    let app = app.fallback(on(
        MethodFilter::GET
            | MethodFilter::HEAD
            | MethodFilter::OPTIONS
            | MethodFilter::PATCH
            | MethodFilter::POST
            | MethodFilter::PUT
            | MethodFilter::TRACE,
        handler_404,
    ));

    match ip {
        Some(v) => {
            let host: Vec<u8> = v.split(".").map(|s| s.parse::<u8>().unwrap()).collect();
            let addr = SocketAddr::from(([host[0], host[1], host[2], host[3]], port));
            info!("Listening on  {addr}");
            axum::Server::bind(&addr)
                .serve(app.into_make_service())
                .await
                .unwrap();
        }
        None => {
            let host: Vec<u8> = local_ipaddress::get()
                .unwrap()
                .split(".")
                .map(|s| s.parse::<u8>().unwrap())
                .collect();

            let addr = SocketAddr::from(([host[0], host[1], host[2], host[3]], port));
            info!("Listening on  {addr}");
            axum::Server::bind(&addr)
                .serve(app.into_make_service())
                .await
                .unwrap();
        }
    };
}

fn init_kafka() {
    // let producer = Producer::from_hosts(vec!["localhost:9092".to_owned()])
    //     .with_ack_timeout(Duration::from_secs(1))
    //     .with_required_acks(RequiredAcks::One)
    //     .create();
    // match producer {
    //     Ok(mut p) => {
    //         let mut buf = String::with_capacity(2);
    //         // for i in 0..1000 {
    //         let _ = write!(
    //             &mut buf,
    //             "{}",
    //             "10000,2017-10-01,北京,20,0,2017-10-01 06:00:00,20,10,10"
    //         ); // some computation of the message data to be sent
    //         let res = p.send(&Record::from_value("my-topic", buf.as_bytes()));
    //         match res {
    //             Ok(_) => {
    //                 info!("send success");
    //             }
    //             Err(e) => {
    //                 info!("send error {}", e);
    //             }
    //         }

    //         buf.clear();
    //         // }
    //     }
    //     Err(e) => {
    //         info!("error: {}", e);
    //     }
    // }

    tokio::spawn(async move {
        let mut consumer = Consumer::from_hosts(vec!["localhost:9092".to_owned()])
            .with_topic_partitions("my-topic".to_owned(), &[0])
            .with_fallback_offset(FetchOffset::Latest)
            .with_group("my-group".to_owned())
            .with_offset_storage(GroupOffsetStorage::Kafka)
            .create()
            .unwrap();
        loop {
            for ms in consumer.poll().unwrap().iter() {
                for m in ms.messages() {
                    info!("receive message: {:?}", m);
                }
                consumer.consume_messageset(ms).unwrap();
            }
            consumer.commit_consumed().unwrap();

            tokio::time::sleep(Duration::from_secs(15)).await;
        }
    });
}
